/**
 * 
 */
package edu.cmu.vlis.wassup.publisher;

import java.io.BufferedWriter;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import edu.cmu.vlis.wassup.CoordinatorResponder;
import edu.cmu.vlis.wassup.IProcess;

import sun.misc.Signal;

/**
 * This is an abstract poller which all poller must inherit.
 */
public abstract class AbstractPoller implements Runnable, IProcess{

	TopicSession topicSession;
	TopicPublisher topicPublisher;
	TopicConnection topicConn;
	CoordinatorResponder cr;
	boolean stop;
	String name;
	protected  BufferedWriter out1;
	protected String filename = "test";

	public AbstractPoller(int id) {
		//register hook method
		Runtime.getRuntime().addShutdownHook(new
				ShutdownPoller(topicConn));
		filename = "LogU"+id;
		FileWriter f;
		try {
			f = new FileWriter(filename);
			out1 = new BufferedWriter(f);
			loadProperties();
			TopicConnectionFactory topicConnFactory = new com.sun.messaging.TopicConnectionFactory();
			try {
				String jmsURL = InetAddress.getLocalHost().getHostAddress()+":7676";    //Queues will always be on localhost.
				((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
						com.sun.messaging.ConnectionConfiguration.imqAddressList, jmsURL);
				((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
						com.sun.messaging.ConnectionConfiguration.imqDefaultUsername, "admin");
				((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
						com.sun.messaging.ConnectionConfiguration.imqDefaultPassword, "admin");

				this.topicConn= topicConnFactory.createTopicConnection();
				this.topicSession = this.topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
				this.topicPublisher = this.topicSession.createPublisher(null); 
				out1.write("Init done!\n");
				out1.flush();

			} catch (JMSException e) {
				// TODO Auto-generated catch block
				out1.write("Init Exc1!\n");
				out1.flush();
				e.printStackTrace();
			} catch (UnknownHostException e) {
				// TODO Auto-generated catch block
				out1.write("Init Exc2!\n");
				out1.flush();
				e.printStackTrace();
			}
			out1.write("In constructor\n");
			out1.flush();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		stop = false;
	}

	public void start(String name){

		Thread pollerThread = new Thread(this);
		pollerThread.start();
		this.name = name;        
		Thread crThread = new Thread(new CoordinatorResponder(this.getID()));
		crThread.start();
	}   

	public void run(){
		try {
			out1.write("run Called\n");
			out1.flush();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		this.startPoller();
	}

	private void loadProperties(){
		Properties props = new Properties();
		try {
			props.load(new FileInputStream("src/properties"));
			System.getProperties().putAll(props);

		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	protected void  startPoller(){
		try {
			topicConn.start();
			try {
				out1.write("Start Called\n");
				out1.flush();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			stop = false;
			while(!stop) {  
				//Keep polling after 10 seconds
				try {
					try {
						out1.write("Calling MsgInget\n");
						out1.flush();
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					int msgPolled = getMessagesNputInPublisherQ();
					System.out.printf("%s polled %d messages.", getPollerName(),msgPolled);
					System.out.println();
					out1.write("%s polled %d messages." + getPollerName()+ " " +msgPolled);
					out1.write("out of  get !\n");
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		} catch (JMSException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	}

	public void stop()  {
		stop = true;
		try {
			if (topicConn != null)
				topicConn.stop();
			System.out.println("Stoped the Poller");
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void shutDown() {
		System.out.println("Sending Int signal");
		Signal.raise(new Signal("INT"));
	}
	/**
	 * Function which actually does the polling.
	 * @return returns number of messages polled.
	 */
	public abstract int getMessagesNputInPublisherQ();

	public abstract String getPollerName();

	public abstract int getID();

}
